package cn.fan.vertd.verticle;

import cn.fan.vertd.utils.TunerKafka;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.json.JsonObject;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.time.Duration;

public class KafkaConsumerVerticle extends AbstractVerticle {

    private static final Logger log = LogManager.getLogger();

    final TunerKafka kafka = new TunerKafka();

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        init(config())
                .compose(this::subscribeTestTopic)
                .onSuccess(startPromise::complete)
                .onFailure(startPromise::fail);
    }

    @Override
    public void stop(Promise<Void> stopPromise) throws Exception {
        super.stop(stopPromise);
    }

    private Future<Void> init(JsonObject conf) {
        Promise<Void> promise = Promise.promise();
        kafka.createConsumer(vertx, conf, TunerKafka.TEST_GROUP_ID);
        promise.complete();
        return promise.future();
    }

    private Future<Void> subscribeTestTopic(Void unused) {
        return kafka.getConsumer()
                .subscribe(TunerKafka.TEST_TOPIC)
                .onSuccess(event -> {
                    log.error("subscribe topic [{}] succeed", TunerKafka.TEST_TOPIC);
                    vertx.setPeriodic(500, timeId -> {
                        kafka.getConsumer().poll(Duration.ofMillis(10)).onSuccess(records -> {
                            log.info("recived message: {}", records.size());
                        });
                    });
                })
                .onFailure(event -> {
                    log.info("error: ", event);
                });
    }
}
